package spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_item_allot;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import spring.cloud.tasks.common.node.BlockNodePaths;
import spring.cloud.tasks.common.task_item_allot.OnlineExecutor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Setter
@Getter
public class TaskItemAllotContentService {

    public static final String GET_SHARD_FROM_CONTENT_NODE_FLAG = "get_shard_from_content_node_flag";
    private CuratorFramework curatorFramework;

    private Gson gson = new Gson();

    public TaskItemAllotContentService(CuratorFramework curatorFramework) {
        this.curatorFramework = curatorFramework;
    }

    public OnlineExecutor getOnlineExecutor(String executorId) throws Exception {
        // Sharding/content 内容多的时候，分多个节点存数据
        if (curatorFramework.checkExists().forPath(BlockNodePaths.ONLINE_EXECUTOR_TASK_ITEM_ASSIGNS_NODE_PATH) != null) {
            List<String> childrenPathList = curatorFramework.getChildren().forPath(BlockNodePaths.ONLINE_EXECUTOR_TASK_ITEM_ASSIGNS_NODE_PATH);
            Collections.sort(childrenPathList, new Comparator<String>() {
                @Override
                public int compare(String arg0, String arg1) {
                    Integer a = Integer.valueOf(arg0);
                    Integer b = Integer.valueOf(arg1);
                    return a.compareTo(b);
                }
            });
            for (String childrenPath : childrenPathList) {
                if (executorId.equals(childrenPath)) {
                    byte[] bytes = curatorFramework.getData().forPath(BlockNodePaths.getOnlineExecutorTaskItemAssignsNodePath(childrenPath));
                    OnlineExecutor onlineExecutor = gson.fromJson(new String(bytes, StandardCharsets.UTF_8), new TypeToken<OnlineExecutor>() {
                    }.getType());
                    return onlineExecutor;
                }
            }
            return null;
        } else {
            return null;
        }
    }

    public ConcurrentHashMap<String, ConcurrentHashMap<String, List<Integer>>> getTaskIdToTaskInstanceIdToItemIndexListMapMap(OnlineExecutor onlineExecutor) {
        ConcurrentHashMap<String, ConcurrentHashMap<String, List<Integer>>> taskIdToTaskInstanceIdToItemIndexListMapMap = new ConcurrentHashMap<>();
        List<TaskItem> taskItemList = onlineExecutor.getAssignedTaskItemList();
        //
        List<String> taskIdList = new ArrayList<>();
        for (TaskItem taskItem : taskItemList) {
            if (!taskIdList.contains(taskItem.getTaskId())) {
                taskIdList.add(taskItem.getTaskId());
            }
        }
        //获取每个任务下面实例的最新的实例轮次id,即使在分配的时候重复了,在执行的时候也不要重复执行。
        Map<String, Long> map = new HashMap<String, Long>();
        for (TaskItem taskItem : taskItemList) {
            String taskId = taskItem.getTaskId();
            String taskInstanceId = taskItem.getTaskInstanceId();
            String taskRoundId = taskItem.getTaskRoundId();
            long dateTime = Long.parseLong(taskRoundId.split("#")[1]);
            if (!map.containsKey(taskInstanceId)) {
                map.put(taskInstanceId, dateTime);
            } else {
                long maxDateTime = map.get(taskInstanceId);
                if (maxDateTime < dateTime) {
                    map.put(taskInstanceId, dateTime);
                }
            }
        }

        for (TaskItem taskItem : taskItemList) {
            String taskId = taskItem.getTaskId();
            String taskInstanceId = taskItem.getTaskInstanceId();
            String taskRoundId = taskItem.getTaskRoundId();
            long dateTime = Long.parseLong(taskRoundId.split("#")[1]);
            long maxDateTime = map.get(taskInstanceId);
            if (dateTime == maxDateTime) {
                taskIdToTaskInstanceIdToItemIndexListMapMap.putIfAbsent(taskId, new ConcurrentHashMap<>());
                ConcurrentHashMap<String, List<Integer>> taskInstanceIdToItemIndexListMap = taskIdToTaskInstanceIdToItemIndexListMapMap.get(taskId);
                taskInstanceIdToItemIndexListMap.putIfAbsent(taskInstanceId, new ArrayList<Integer>());
                List<Integer> itemIndexList = taskInstanceIdToItemIndexListMap.get(taskId);
                itemIndexList.add(taskItem.getTaskItemIndex());
            }
        }
        return taskIdToTaskInstanceIdToItemIndexListMapMap;
    }
}
